package com.myz.rabbitmq.autoAck;

import com.myz.rabbitmq.utils.ConnectionUtils;
import com.rabbitmq.client.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.concurrent.TimeoutException;

/**
 * 消息消费者2
 *
 * @author maoyz on 18-4-4.
 */
public class Work2 {
    private static final Logger logger = LoggerFactory.getLogger(Work2.class);
    private static final String TASK_QUEUE_NAME = "work_queue";

    public static void main(String[] args) throws IOException, TimeoutException {
        Connection connection = ConnectionUtils.getConnection();
        final Channel channel = connection.createChannel();

        channel.queueDeclare(TASK_QUEUE_NAME, true, false, false, null);

        logger.debug("Worker2  Waiting for messages");

        channel.basicQos(1);
        // 定义消费者
        final Consumer consumer = new DefaultConsumer(channel) {

            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                String message = new String(body, "UTF-8");

                logger.debug("Worker2  Received '" + message + "'");

                try {
                    // throw new Exception();
                    doWork(message);
                } catch (Exception e) {
                    channel.abort();
                } finally {
                    logger.debug("Worker2 Done");
                    channel.basicAck(envelope.getDeliveryTag(), false);
                }
            }
        };
        boolean autoAck = false;
        //消息消费完成确认
        channel.basicConsume(TASK_QUEUE_NAME, autoAck, consumer);
    }

    private static void doWork(String task) {
        try {
            Thread.sleep(2000); // 暂停1秒钟
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}
